package com.udf.flink.scala.apitest.state

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, MapState, MapStateDescriptor, ValueState}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector
//import org.apache.flink.streaming.api.scala._
import scala.collection.JavaConverters._


object MapStateEx {
  def main(args: Array[String]): Unit = {
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    // Checkpoint
    streamEnv.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE)
    streamEnv.setStateBackend(new HashMapStateBackend());
    streamEnv.getCheckpointConfig.setCheckpointStorage("file:///home/xuzh");
    streamEnv.socketTextStream("localhost", 9100)
      .setParallelism(1)
      .map(x => (x.split(":")(0), x.split(":")(1).toInt:Integer))
      .keyBy(_._1)
      .process(new ListStaeFunc())
      .print()
    streamEnv.execute("list")
  }
}
class MapStaeFunc extends KeyedProcessFunction[String,(String,Integer),(String,List[String])]{
  private var listState: MapState[String,Integer] = _
  override def open(parameters: Configuration): Unit ={
    val listStatDesc =new MapStateDescriptor[String,Integer]("MapStateDesc",
      TypeInformation.of(classOf[String]), TypeInformation.of(classOf[Integer]))
    listState = getRuntimeContext.getMapState(listStatDesc);
  }
  override def processElement(value: (String, Integer), ctx: KeyedProcessFunction[String, (String, Integer), (String, List[String])]#Context, out: Collector[(String, List[String])]): Unit = {
//       val tp =value._2
//      listState.add(tp)
////    val iterable = listState.get().iterator();
//      val charList: List[Integer] = listState.get().asScala.toList
//      out.collect(value._1,charList)
    listState.put(value._1,value._2)
//    val rs :Map[String,Integer]

    out.collect((value._1,listState.keys().asScala.toList))
  }
}

//https://blog.csdn.net/winterking3/article/details/115086744?utm_medium=distribute.pc_aggpage_search_result.none-task-blog-2~aggregatepage~first_rank_ecpm_v1~rank_v31_ecpm-2-115086744.pc_agg_new_rank&utm_term=flink%E7%9A%84mapstate%E4%BE%8B%E5%AD%90&spm=1000.2123.3001.4430
